package org.niit.sql

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object SparkSQL_1 {

  def main(args: Array[String]): Unit = {
    //准备环境
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQL")
    val spark: SparkSession = SparkSession.builder().config(sparkConf).getOrCreate()
    spark.sparkContext.setLogLevel("ERROR")
    //1.DataFrame
    //1.1使用spark SQL 读取 json文件
    val dataFrame: DataFrame = spark.read.json("input/user.json")
    //1.2展示DataFrame信息
    dataFrame.show()
    //1.3想使用SQL进行查询，需要使用dataFrame去创建一张临时表，并起名字。后续对临时表进行查询即可
    dataFrame.createOrReplaceTempView("user")
    //1.4 使用临时表进行查询
    val resDF1: DataFrame = spark.sql("select * from user")
    resDF1.show()
    //1.5 查看DataFrame结构
    dataFrame.printSchema()
    //1.6只查看 username列 数据
    var resDF2: DataFrame = spark.sql("select username from user")
    resDF2 = dataFrame.select("username")
    resDF2.show()
    //1.7查看 username列数据 以及 "age + 1" 数据
    //如果需要使用DSL 需要引入 spark.implicits._  DSL:去管理结构化的数据特定领域的语言。使用DSL不用创建临时视图
    import spark.implicits._
    dataFrame.select($"username",$"age"+1 ).show()
    dataFrame.select('username,'age +1 as "newAge" ).show()

    //1.8查看 age 大于 30的数据
    dataFrame.filter($"age" >30).show()
    //1.9按照age 分组，查看数据的条数
    dataFrame.groupBy("age").count().show()
    //1.10 RDD 转换 DataFrame
    val sparkContext: SparkContext = spark.sparkContext
    val rdd1: RDD[Int] = sparkContext.makeRDD((1 to 4).toList)
    val df1: DataFrame = rdd1.toDF("id")
    df1.show()
    //1.11 DataFrame 转换成 RDD    toDF  rdd   DataFrame = RDD[ROW]
    val rdd2: RDD[Row] = df1.rdd   // df1 == df1.toDF  目的都是要去掉表头[结构化约束]
    val array: Array[Row] = rdd2.collect()
    array.foreach( println )

    //2. DataSet 是具有强类型的集合。需要提供对应的类型信息
    //2.1使用样例类创建DataSet
    val list = List( Person("zhangsan",40),Person("lisi",50) )
    val ds1: Dataset[Person] = list.toDS()
    ds1.show()
    //2.2使用基本数据类型创建DataSet
    val ds2: Dataset[Int] = List(1, 2, 3, 4, 5).toDS()
    ds2.show()
    //2.3 DataSet转换 RDD
    val rdd3: RDD[Person] = ds1.rdd
    rdd3.collect().foreach( println )

    //2.4RDD 转换 DataSet
    val ds3: Dataset[Person] = rdd3.toDS()
    ds3.show()

    //特殊的RDD 转成DataSet
    val rdd: RDD[(String, Int)] = sparkContext.makeRDD(List( ("wangwu", 50), ("maliu", 60)))
    val ds5: Dataset[Person] = rdd.map {
      case (name, age) => {
        Person(name, age)
      }
    }.toDS()
    ds5.show()

    //2.5DataFrame和 DataSet相互转换
    val df3: DataFrame = ds1.toDF()
    df3.show()

    val ds4: Dataset[Person] = df3.as[Person]
    ds4.show()
    //关闭Spark SQL
    spark.close()
  }
  //样例类
  case class Person(name:String,age:Long)
}
